﻿using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.Linq;
using System.Text;

namespace Universe.SimpleQueue
{
    partial class SimpleQueue
    {

        public Message NextDelivery(string queueName)
        {
            var connectionInfo = ConnectionFactory();
            var con = connectionInfo.Connection;
            var tran = connectionInfo.Transaction;
            DumpConnection(con, tran);

            try
            {

                String sql1 =
                    "SELECT Top 1 Id FROM SimpleQueue WITH (UPDLOCK, ROWLOCK) Where " +
                    " (QueueName = @queueName) " +
                    " AND (Locked = 0) " +
                    " AND ((DeliveryDate is null) or (DeliveryDate >= @now)) " +
                    " AND (AckDate is null) " +
                    " ORDER BY ModifiedAt";

                string id;
                var cmd = AdoHelper.BindParameters(con, tran, sql1, new { queueName, now = UniversalTime });
                using (cmd)
                {
                    id = (string)cmd.ExecuteScalar();
                    if (id == null)
                    {
                        if (tran != null) tran.Commit();
                        DumpConnection(con, tran);
                        return null;
                    }
                }

                const string sql2 = "Update SimpleQueue WITH (ROWLOCK) Set Locked = 1, HandlersCount = HandlersCount + 1 Where Id = @Id";
                var cmd2 = AdoHelper.BindParameters(con, tran, sql2, new { id, });
                using (cmd2)
                    cmd2.ExecuteNonQuery();

                const string sql3 = "Select * From SimpleQueue WITH (ROWLOCK) Where Id = @Id ";
                var cmd3 = AdoHelper.BindParameters(con, tran, sql3, new { id, });
                using (cmd3)
                {
                    Message ret = null;
                    using (var rdr3 = cmd3.ExecuteReader(CommandBehavior.SingleRow))
                    {
                        if (rdr3.Read())
                        {
                            ret = new Message
                            {
                                AckDate = AdoHelper.ReadDate(rdr3, "AckDate"),
                                CreatedAt = AdoHelper.ReadDate(rdr3, "CreatedAt"),
                                Id = AdoHelper.ReadString(rdr3, "Id"),
                                Locked = AdoHelper.ReadBoolean(rdr3, "Locked"),
                                ModifiedAt = AdoHelper.ReadDate(rdr3, "ModifiedAt"),
                                QueueName = AdoHelper.ReadString(rdr3, "QueueName"),
                                OptionalKey = AdoHelper.ReadString(rdr3, "OptionalKey"),
                                Payload = AdoHelper.ReadBinary(rdr3, "Payload"),
                                HandlersCount = AdoHelper.ReadInt32(rdr3, "HandlersCount").Value,
                            };
                        }
                    }

                    if (tran != null) tran.Commit();
                    DumpConnection(con, tran);
                    LocalStat.IncInProcess(queueName, 1);
                    return ret;
                }
            }
            catch (Exception ex)
            {
                if (tran != null)
                    tran.Rollback();

                SqlException sex = ex as SqlException;
                bool isDeadLock = (sex != null && sex.Number == 1205);
                bool isCeLockTimeout = ex.GetType().Name.Equals("SqlCeLockTimeoutException");

                throw new InvalidOperationException("NextDelivery() from the queue '" + queueName + "' failed", ex);
            }
            finally
            {
                DumpConnection(con, tran);
                con.Close();
            }
        }

        public void Ack(String queueName, string idMessage)
        {

            var connectionInfo = ConnectionFactory();
            var con = connectionInfo.Connection;
            var tran = connectionInfo.Transaction;

            DumpConnection(con, tran);

            try
            {
                String sql = "Update SimpleQueue WITH (ROWLOCK) Set AckDate = @now, Locked = 0 Where Id = @Id";
                var cmd = AdoHelper.BindParameters(con, tran, sql, new { now = UniversalTime, id = idMessage });
                using (cmd)
                    cmd.ExecuteNonQuery();

                if (tran != null) tran.Commit();
                LocalStat.IncAck(queueName);
                LocalStat.IncInProcess(queueName, -1);
            }
            catch (Exception ex)
            {
                if (tran != null)
                    tran.Rollback();

                throw new InvalidOperationException(
                    "Ack() of the message '" + idMessage + "' in the queue '" + queueName + "' failed",
                    ex);
            }
            finally
            {
                DumpConnection(con, tran);
                con.Close();
            }

        }

        public void Postpone(String queueName, String idMessage, DateTime deliveryAtUtc = default(DateTime))
        {

            if (deliveryAtUtc >= new DateTime(1, 1, 1) && deliveryAtUtc.Year < 1901)
                throw new ArgumentException("deliveryAtUtc parameter requires Year >= 1901");

            var connectionInfo = ConnectionFactory();
            var con = connectionInfo.Connection;
            var tran = connectionInfo.Transaction;

            DumpConnection(con, tran);

            try
            {
                const string sql = "Update SimpleQueue WITH (ROWLOCK) Set DeliveryDate = @deliveryDate, Locked = 0, ModifiedAt = @now Where Id = @id";
                var parameters = new
                {
                    @deliveryDate = deliveryAtUtc == default(DateTime) ? (DateTime?)null : (DateTime?)deliveryAtUtc,
                    now = UniversalTime,
                    Id = idMessage
                };

                var cmd = AdoHelper.BindParameters(con, tran, sql, parameters);

                using (cmd)
                    cmd.ExecuteNonQuery();

                if (tran != null) tran.Commit();
                LocalStat.IncInProcess(queueName, -1);
                LocalStat.IncPostpone(queueName);
            }
            catch (Exception ex)
            {
                if (tran != null)
                    tran.Rollback();

                throw new InvalidOperationException(
                    "Postpone() of the message '" + idMessage + "' in the queue '" + queueName + "' failed",
                    ex);
            }
            finally
            {
                DumpConnection(con, tran);
                con.Close();
            }
        }

    }
}
